package com.wujian.love.rocketmq.consumer;

import com.wujian.love.contants.RocketMQContants;
import com.wujian.love.domin.OrderParent;
import com.wujian.love.mapper.OrderParentMapper;
import com.wujian.love.utils.JsonSerilizable;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.Map;

/**
 * @ClassName: Consumer
 * @description:
 * @author: wujian
 * @Date: 2020-07-07 16:59
 **/
@Component
public class OrderConsumer {

    private static volatile SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    private DefaultMQPushConsumer consumer;

    private static final String CONSUMER_GROUP_NAME = "Love_Test_Consumer_Order_GroupName";


    @Autowired
    private OrderParentMapper orderParentMapper;

    private OrderConsumer() {
        try {
            this.consumer = new DefaultMQPushConsumer(CONSUMER_GROUP_NAME);
            this.consumer.setConsumeThreadMin(10);
            this.consumer.setConsumeThreadMax(30);
            this.consumer.setNamesrvAddr(RocketMQContants.NAMESERVER);
            this.consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            this.consumer.subscribe(RocketMQContants.ORDER_TOPIC, RocketMQContants.ORDER_CANCEL_TAG);
            this.consumer.registerMessageListener(new MessageListenerConcurrently4Pay());
            this.consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    class MessageListenerConcurrently4Pay implements MessageListenerConcurrently {

        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            System.out.println("延时消息消费时间:"+sdf.format(new Date()));
            MessageExt msg = msgs.get(0);
            try {
                String topic = msg.getTopic();
                String tags = msg.getTags();
                String keys = msg.getKeys();
                String body = new String(msg.getBody(), RemotingHelper.DEFAULT_CHARSET);
                System.err.println("收到延时事务消息, topic: " + topic + ", tags: " + tags + ", keys: " + keys + ", body: " + body);

                // 消息一单过来的时候（去重 幂等操作）
                // 数据库主键去重<去重表 keys>
                //     insert table --> insert ok & primary key

                Map<String, Object> paramsBody = JsonSerilizable.deserilizableForMapFromFile(body, Object.class);
                Integer orderParentId = (Integer) paramsBody.get("orderParentId");    //     统一的订单
                OrderParent orderParent = orderParentMapper.selectOrderParentById(orderParentId);
                System.out.println(orderParentId);
                if(orderParent != null && orderParent.getOrderStatus().equals(OrderParent.WAIT_ORDER_PAY)){
                    System.out.println("订单未支付,需要关闭订单");
                    System.out.println("consumeMessage : 延时消息->取消订单->订单id"+orderParentId);
                }

            } catch (Exception e) {
                e.printStackTrace();
                //msg.getReconsumeTimes();
                // 如果处理多次操作还是失败, 记录失败日志（做补偿 回顾 人工处理）
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }


    }
}